package com.sf.osms.dep.message;

import java.util.List;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import com.sf.fvp.ConvertUtil;
import com.sf.fvp.dto.FactRouteDto;
import com.sf.kafka.api.consume.ConsumeConfig;
import com.sf.kafka.api.consume.IByteArrayMessageConsumeListener;
import com.sf.kafka.api.consume.KafkaConsumeRetryException;
import com.sf.kafka.api.consume.KafkaConsumerRegister;
import com.sf.osms.dep.dao.IRouterDao;

@Component
public class KafkaRouterMessageListener implements IByteArrayMessageConsumeListener , InitializingBean{

	private Logger logger = LoggerFactory.getLogger(getClass());
	
	@Value("${dep.kafka.route.monitorUrl}")
	private String monitorUrl = null;
	@Value("${dep.kafka.route.clusterName}")
	private String clusterName = null;
	@Value("${dep.kafka.route.systemIdToken}")
	private String systemIdToken = null;
	@Value("${dep.kafka.route.topic}")
	private String topic = null;
	@Value("${dep.kafka.route.consumeThreadCount}")
	private int threadCount = 5;
	@Value("${dep.kafka.route.active}")
	private boolean active; // 是否启用
	@Autowired
	private IRouterDao routerDao;
	private ConsumeConfig consumeConfig = null;
	@Override
	public void onMessage(List<byte[]> messages) throws KafkaConsumeRetryException {
	    for (byte[] bs : messages) {
	    	FactRouteDto frd = ConvertUtil.fromByte(FactRouteDto.class, bs); 
	    	String sfwaybillNo = frd.getMainWaybillNo();
	    	int status = frd.getStatus() + 10;
	    	Integer eventCode = null;
	    
	    	if(StringUtils.isNotEmpty(frd.getOpCode())){
	    		eventCode = Integer.valueOf(frd.getOpCode().trim());
	    	}
	    	routerDao.updateRouterBySfWaybill(sfwaybillNo, eventCode,status);
	    }
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		 if ((StringUtils.isNotEmpty(this.monitorUrl)) && (StringUtils.isNotEmpty(this.systemIdToken)) && 
			 (StringUtils.isNotEmpty(this.clusterName)) && (StringUtils.isNotEmpty(this.topic))){
			 	  if(active){
			 		 this.consumeConfig = new ConsumeConfig(this.systemIdToken, this.monitorUrl, this.clusterName, this.topic, this.threadCount);
				     logger.info("Kafka consumer with clusterName : {} topic : {} systemIdToken : {}, config status ： valid.", new Object[] { this.clusterName, this.topic, this.systemIdToken });
				     KafkaConsumerRegister.registerByteArrayConsumer(this.consumeConfig,this); 
				      
			 	  }
			     
			      
		}
	}

}
